package com.xtayfjpk.zookeeper.test;

import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class Executor implements Watcher, Runnable, DataMonitorListener {
	String znode;
	DataMonitor dm;
	ZooKeeper zk;
	String filename;
	String exec[];
	Process child;
	
	public static void main(String[] args) throws Exception {
		String hostPort = "192.168.0.202:2181";
		String znode = "/zk_test";
		String filename = "H:/zk_test.txt";
		String[] exec = new String[0];
		//new Executor(hostPort, znode, filename, exec).run();
		
		ZooKeeper zk = new ZooKeeper(hostPort, 3000, new Watcher() {
			
			@Override
			public void process(WatchedEvent event) {
				System.out.println(event);
			}
		});
		
		registerWatcher(znode, zk);
		zk.exists(znode, true, new StatCallback() {
			
			@Override
			public void processResult(int rc, String path, Object ctx, Stat stat) {
				System.out.println(path);	
				System.out.println(rc);
				System.out.println(stat);
			}
		}, null);
		
		byte[] data = zk.getData(znode, false, null);
		System.out.println(new String(data));
		
		
		Object object = new Object();
		synchronized (object) {
			object.wait();
		}
	}


	private static void registerWatcher(final String znode, final ZooKeeper zk)
			throws KeeperException, InterruptedException {
		zk.exists(znode, new Watcher() {
			
			@Override
			public void process(WatchedEvent event) {
				System.out.println("zk_test: " + event);
				
				try {
					registerWatcher(znode, zk);
				} catch (KeeperException | InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		});
	}
	
	
	public Executor(String hostPort, String znode, String filename,
            String exec[]) throws KeeperException, IOException {
        this.filename = filename;
        this.exec = exec;
        zk = new ZooKeeper(hostPort, 3000, this);
        dm = new DataMonitor(zk, znode, null, this);
    }
	
	
	@Override
	public void run() {
		try {
			synchronized (this) {
				while(!dm.dead) {
					wait();
				}
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
	}

	@Override
	public void process(WatchedEvent event) {
		dm.process(event);		
	}

	

	@Override
	public void closing(int rc) {
		synchronized (this) {
			while(!dm.dead) {
				notifyAll();
			}
		}		
	}

	
	static class StreamWriter extends Thread {
		OutputStream os;
		InputStream is;
		
		StreamWriter(InputStream is, OutputStream os) {
			this.is = is;
			this.os = os;
			start();
		}
		
		@Override
		public void run() {
			byte[] b = new byte[80];
			int rc;
			try {
				while((rc=is.read())>0) {
					os.write(b, 0, rc);
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		
	}
	
	@Override
	public void exists(byte[] data) {
		if(data==null) {
			if(child!=null) {
				System.out.println("Killing process");
				child.destroy();
				try {
					child.waitFor();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			child = null;
		} else {
			if(child!=null) {
				System.out.println("Stopping child");
				child.destroy();
				try {
					child.waitFor();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			
			try {
				FileOutputStream fos = new FileOutputStream(filename);
				fos.write(data);
				fos.close();
			} catch(IOException e) {
				e.printStackTrace();
			}
			
			try {
				System.out.println("Starting child");
				child = Runtime.getRuntime().exec(exec);
			} catch(IOException e) {
				e.printStackTrace();
			}
		}
	}
}
